package main

import (
	"context"
	"gitee.com/lihaiping1603/utils/log"
	"gitee.com/lihaiping1603/utils/xhttp"
	"github.com/gin-gonic/gin"
	"io"
	"net/http"
	"os"
	"os/signal"
	"strings"
	"syscall"
)

type ICache interface {
	Name() string
	Buckup(key string) error
	io.Reader // Reader must continue reading after EOF on subsequent calls after more Writes.
	io.Writer // Concurrent reading/writing must be supported.
	io.Closer // Close should do any cleanup when done with the File
}

type StreamCache struct {
	name string
	pr   *io.PipeReader
	pw   *io.PipeWriter
}

func NewStreamCache(name string) *StreamCache {
	pr, pw := io.Pipe()
	return &StreamCache{pr: pr, pw: pw, name: name}
}

func (sc *StreamCache) Name() string {
	return sc.name
}

func (sc *StreamCache) Buckup(key string) error {
	return nil
}

func (sc *StreamCache) Read(p []byte) (n int, err error) {
	return sc.pr.Read(p)
}

func (sc *StreamCache) Write(data []byte) (n int, err error) {
	return sc.pw.Write(data)
}

func (sc *StreamCache) Close() error {
	sc.pw.Close()
	return nil
}

type HttpSender struct {
	client   *http.Client
	endpoint string
}

func NewHttpSender(client *http.Client) *HttpSender {
	return &HttpSender{client: client}
}

func (sender *HttpSender) DoSend(ctx context.Context, url string, method string, body io.Reader, fc func(r *http.Request)) (*http.Response, error) {
	request, err := http.NewRequestWithContext(ctx, method, url, body)
	if err != nil {
		log.Error("client send to url:%s,err:%s", url, err.Error())
		return nil, err
	}
	if fc != nil {
		fc(request)
	}
	log.Info("client send to url:%s,header:%+v", url, request.Header)
	return sender.client.Do(request)
}

func writeResponse(c *gin.Context, err error) {
	if err != nil {
		c.AbortWithError(http.StatusInternalServerError, err)
	} else {
		c.AbortWithStatus(http.StatusOK)
	}
	return
}

func copyHeaders2(dst, src http.Header, keep_dest_headers bool) {
	if !keep_dest_headers {
		for k := range dst {
			dst.Del(k)
		}
	}
	for k, vs := range src {
		kk := strings.ToLower(k)
		switch kk {
		case "etag":
			k = "ETag"
		case "x-amz-request-id":
			k = "x-amz-request-id"
		case "x-amz-bucket-owner":
			k = "x-amz-bucket-owner"
		case "x-amz-op-type":
			k = "x-amz-op-type"
		case "x-amz-user-account":
			k = "x-amz-user-account"
		}
		dst[k] = vs
	}
}

func relayHandle(c *gin.Context, rurl string, stream_cache ICache) {
	w := c.Writer
	//需要注意，这个地方的rurl中，是由“endpoint+uri”构成的了
	res, err := relayToServer(c, rurl, stream_cache)
	if err != nil { //转发失败
		log.Error("relay to server:%s err:%s", rurl, err.Error())
		writeResponse(c, err)
		return
	}
	//关闭body
	defer func() {
		if res.Body != nil {
			res.Body.Close()
		}
	}()
	//将转发的回复，再写回给请求方
	copyHeaders2(w.Header(), res.Header, false)
	c.Status(res.StatusCode)
	if res.Body != nil {
		if nr, err := io.Copy(w, res.Body); err != nil {
			log.Error("io copy body:%d err:%s", nr, err.Error())
		}
	}
	log.Info("write response:%d,header:%+v", res.StatusCode, w.Header())
}

func relayToServer(c *gin.Context, rurl string, stream_cache ICache) (*http.Response, error) {
	sender := NewHttpSender(http.DefaultClient)
	res, err := sender.DoSend(context.Background(), rurl, http.MethodPut, stream_cache, func(request *http.Request) {
		request.Header = c.Request.Header.Clone()
		request.Host = c.Request.Host
		//Expect 是一个请求消息头，包含一个期望条件，表示服务器只有在满足此期望条件的情况下才能妥善地处理请求。
		//规范中只规定了一个期望条件，即 Expect: 100-continue, 对此服务器可以做出如下回应：
		//100 如果消息头中的期望条件可以得到满足，使得请求可以顺利进行的话
		request.Header.Del("Expect")
	})
	if err != nil {
		log.Error("send to server:%s err:%s", rurl, err.Error())
		return nil, err
	}
	if res.StatusCode != http.StatusOK {
		log.Warn("remote server response code:%d,header:%+v", res.StatusCode, res.Header)
		return res, nil
	}
	log.Info("remote server response code:%d,header:%+v", res.StatusCode, res.Header)
	return res, nil
}

func streamProxy(c *gin.Context) {
	uri := c.Request.RequestURI
	log.Info("receive uri:%s", uri)
	rs := make(chan struct{})
	cache := NewStreamCache("cache")
	go func() {
		relayHandle(c, "http://rgw.lihaiping.com:8000"+uri, cache)
		log.Info("relay end...")
		cache.Close()
		close(rs)
	}()

	if nr, err := io.Copy(cache, c.Request.Body); err != nil {
		log.Error("io copy data(already reading:%d) to cache err:%s", nr, err.Error())
		c.String(http.StatusBadGateway, err.Error())
		return
	}
	log.Info("org stream write cache end...")
	cache.Close()
	<-rs
}

func register(engine *gin.Engine) {
	proxyRouter := engine.Group("/stream")
	{
		proxyRouter.PUT("/proxy", streamProxy)
	}

}

func main() {
	engine := gin.New()
	engine.Use(gin.Logger(), gin.Recovery())
	server := xhttp.NewServerEngine(engine)
	//注册路由
	register(engine)

	c := make(chan os.Signal, 1)
	signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
	go func() {
		//wait
		<-c
		server.Stop()
		log.Info("stop server...")
	}()

	if err := server.Start(":9090", "", ""); err != nil {
		log.Error("%s", err.Error())
	}
	return
}
